package studio.raptor.ddal.dashboard.log.thread;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.ddal.dashboard.log.common.LogServerConstant;
import studio.raptor.ddal.dashboard.log.queue.interfaces.IMessageQueue;
import studio.raptor.ddal.dashboard.log.queue.interfaces.IQueueManager;
import studio.raptor.ddal.dashboard.log.service.interfaces.ILogStorageService;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

@Component
@Scope("singleton")
public class ThreadManager {

	@Resource
	private IQueueManager queueManager;
	@Autowired
    private  ILogStorageService logStorageService;

	private FrontQueueThread frontQueueOracleThreads[];

	private ExecutorService frontQueueOracleService ;

	private Future[] futureFrontOracleQueues ;

	private FrontQueueThread frontQueueMysqlThreads[];

	private ExecutorService frontQueueMysqlService ;

	private Future[] futureFrontMysqlQueues ;

	/**
	 * 启动线程 用于接收前端接收客户端请求数据并放队列 FRONTSTORAGEQUEUE
	 */
	public void startFrontQueueThread(){
		//根据配置的数据库类型，启动相关数据库对应的队列
		for(int m = 0; m< LogServerConstant.DATABASETYPES.length; m++){
			String databaseType = LogServerConstant.DATABASETYPES[m];
			if(databaseType!=null&&!databaseType.equals("")){
				databaseType = databaseType.toUpperCase();
				if(databaseType.equals(LogServerConstant.DATABASE_ORACLE)){
					IMessageQueue messageQueue = (IMessageQueue)queueManager.getQueue(LogServerConstant.LOG_SERVER_FRONT_DATA_QUEUE_NAME+"_"+databaseType);
					frontQueueOracleService = Executors.newFixedThreadPool(LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER);
					frontQueueOracleThreads = new FrontQueueThread[LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER];
					futureFrontOracleQueues = new Future[LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER];
					for(int i=0;i<LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER;i++){
						frontQueueOracleThreads[i] = new FrontQueueThread(logStorageService,messageQueue);
						futureFrontOracleQueues[i] = frontQueueOracleService.submit(frontQueueOracleThreads[i]);
					}
				}else if(databaseType.equals(LogServerConstant.DATABASE_MYSQL)){
					IMessageQueue messageQueue = (IMessageQueue)queueManager.getQueue(LogServerConstant.LOG_SERVER_FRONT_DATA_QUEUE_NAME+"_"+databaseType);
					frontQueueMysqlService = Executors.newFixedThreadPool(LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER);
					frontQueueMysqlThreads = new FrontQueueThread[LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER];
					futureFrontMysqlQueues = new Future[LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER];
					for(int i=0;i<LogServerConstant.LOG_SERVER_FRONTQUEUETHREAD_NUMBER;i++){
						frontQueueMysqlThreads[i] = new FrontQueueThread(logStorageService,messageQueue);
						futureFrontMysqlQueues[i] = frontQueueMysqlService.submit(frontQueueMysqlThreads[i]);
					}
				}
			}
		}
	}

	/**
	 * 停止线程
	 */
	public void stopFrontQueueThread(){
		if(frontQueueOracleThreads!=null){
			for(int i = 0 ; i < frontQueueOracleThreads.length ; i ++){
				frontQueueOracleThreads[i].setFlag(false);
				futureFrontOracleQueues[i].cancel(true);
				frontQueueOracleThreads[i]=null ;
			}
		}
		if(frontQueueOracleService!=null&&!frontQueueOracleService.isShutdown()){
			frontQueueOracleService.shutdown();
		}
		if(frontQueueMysqlThreads!=null){
			for(int i = 0 ; i < frontQueueMysqlThreads.length ; i ++){
				frontQueueMysqlThreads[i].setFlag(false);
				futureFrontMysqlQueues[i].cancel(true);
				frontQueueMysqlThreads[i]=null ;
			}
		}
		if(frontQueueMysqlService!=null&&!frontQueueMysqlService.isShutdown()){
			frontQueueMysqlService.shutdown();
		}
	}


}
